package org.apache.rocketmq.tools.command;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.tools.admin.MQAdminExt;

import java.util.*;



/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/1/15
 * @方法描述：命令行工具类
 */
public class CommandUtil {

    //Broker集群名称错误提示信息，当指定的集群不存在或连接的Namesrv错误时使用
    private static final String ERROR_MESSAGE = "Make sure the specified clusterName exists or the name server connected to is correct.";

    //没有主节点的时候的占位符
    public static final String NO_MASTER_PLACEHOLDER = "NO_MASTER";


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/15
     * @方法描述：得到指定Broker集群中主节点和从节点映射关系的方法
     */
    public static Map<String/*master addr*/, List<String>/*slave addr*/> fetchMasterAndSlaveDistinguish(final MQAdminExt adminExt, final String clusterName) throws InterruptedException, RemotingConnectException, RemotingTimeoutException, RemotingSendRequestException, MQBrokerException {
        //创建一个map存储主节点和从节点的映射关系
        Map<String, List<String>> masterAndSlaveMap = new HashMap<>(4);
        //获取Broker集群的全部信息
        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
        //根据集群名称，得到该集群下所有Broker节点的名称
        Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
        //判断该Broker集群是否存在
        if (brokerNameSet == null) {
            //不存在则在控制台输出错误信息
            System.out.printf("[error] %s", ERROR_MESSAGE);
            return masterAndSlaveMap;
        }
        //如果存在那就遍历该集群中的每一个Broker名称
        //因为一个Broker集群中可以存在多个master-slave组合
        for (String brokerName : brokerNameSet) {
            //得到该Broker名称对应的BrokerData
            BrokerData brokerData = clusterInfoSerializeWrapper.getBrokerAddrTable().get(brokerName);
            //如果该BrokerData为空或者该BrokerData中的BrokerAddrs为空则跳过本次循环
            if (brokerData == null || brokerData.getBrokerAddrs() == null) {
                continue;
            }
            //得到当前遍历的Broker的master-slave组合中的master节点的地址
            String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
            //如果主节点不存在则使用占位符当作key
            if (masterAddr == null) {
                masterAndSlaveMap.putIfAbsent(NO_MASTER_PLACEHOLDER, new ArrayList<>());
            } else {
                //主节点存在则把主节点地址当作key存放到map中，因为一个主节点可能对应多个从节点，所以value是一个集合，用来存放多个从节点
                masterAndSlaveMap.put(masterAddr, new ArrayList<>());
            }
            //接下来就可以把所有的从节点添加到masterAndSlaveMap这个map中了
            for (Map.Entry<Long, String> brokerAddrEntry : brokerData.getBrokerAddrs().entrySet()) {
                //遍历BrokerData中的BrokerAddrs，如果value为空或者key是master节点则跳过
                if (brokerAddrEntry.getValue() == null || brokerAddrEntry.getKey() == MixAll.MASTER_ID) {
                    continue;
                }
                //这里做了一个判断，那就是判断当前正在处理的Broker主从集群中是否存在主节点
                //如果不存在则把当前的从节点添加到占位符对应的列表中
                if (masterAddr == null) {
                    masterAndSlaveMap.get(NO_MASTER_PLACEHOLDER).add(brokerAddrEntry.getValue());
                } else {
                    //执行到这里就意味着主节点存在，那就直接把从节点添加到与主节点对应的列表中
                    masterAndSlaveMap.get(masterAddr).add(brokerAddrEntry.getValue());
                }
            }
        }
        return masterAndSlaveMap;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/15
     * @方法描述：获取指定Broker集群中所有主节点的方法
     */
    public static Set<String> fetchMasterAddrByClusterName(final MQAdminExt adminExt, final String clusterName)
            throws InterruptedException, RemotingConnectException, RemotingTimeoutException,
            RemotingSendRequestException, MQBrokerException {
        //创建存放主节点地址的集合
        Set<String> masterSet = new HashSet<>();
        //得到Broker集群的全部信息
        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
        //得到制定集群下的所有Broker节点的名称
        Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
        //判断该集群是否存在
        if (brokerNameSet != null) {
            //遍历Broker集群中的每一个Broker名称
            for (String brokerName : brokerNameSet) {
                //得到对应的BrokerData队形
                BrokerData brokerData = clusterInfoSerializeWrapper.getBrokerAddrTable().get(brokerName);
                if (brokerData != null) {
                    //得到主节点的地址
                    String addr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
                    if (addr != null) {
                        //把主节点地址添加到集合中
                        masterSet.add(addr);
                    }
                }
            }
        } else {
            //如果集群不存在则在控制台输出错误信息
            System.out.printf("[error] %s", ERROR_MESSAGE);
        }
        return masterSet;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/15
     * @方法描述：得到指定Broker集群中所有节点地址的方法，主节点和从节点地址都会返回，该方法和上面方法的逻辑基本一致，只是最后返回的节点地址多了一些，所以我就不添加详细注释了
     */
    public static Set<String> fetchMasterAndSlaveAddrByClusterName(final MQAdminExt adminExt, final String clusterName)
            throws InterruptedException, RemotingConnectException, RemotingTimeoutException,
            RemotingSendRequestException, MQBrokerException {
        Set<String> brokerAddressSet = new HashSet<>();
        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
        Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
        if (brokerNameSet != null) {
            for (String brokerName : brokerNameSet) {
                BrokerData brokerData = clusterInfoSerializeWrapper.getBrokerAddrTable().get(brokerName);
                if (brokerData != null) {
                    final Collection<String> addrs = brokerData.getBrokerAddrs().values();
                    brokerAddressSet.addAll(addrs);
                }
            }
        } else {
            System.out.printf("[error] %s", ERROR_MESSAGE);
        }
        return brokerAddressSet;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/15
     * @方法描述：根据Broker名称获取主节点地址的方法
     */
    public static String fetchMasterAddrByBrokerName(final MQAdminExt adminExt,
                                                     final String brokerName) throws Exception {
        //得到Broker集群的全部信息
        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
        //根据Broker的名称得到对应的BrokerData
        BrokerData brokerData = clusterInfoSerializeWrapper.getBrokerAddrTable().get(brokerName);
        if (null != brokerData) {
            //得到主节点地址并返回
            String addr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
            if (addr != null) {
                return addr;
            }
        }
        //如果没有找到对应的Broker信息则抛出异常
        throw new Exception(String.format("No broker address for broker name %s.%n", brokerData));
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/15
     * @方法描述：根据Broker名称获取主节点和从节点地址的方法
     */
    public static Set<String> fetchMasterAndSlaveAddrByBrokerName(final MQAdminExt adminExt, final String brokerName)
            throws InterruptedException, RemotingConnectException, RemotingTimeoutException,
            RemotingSendRequestException, MQBrokerException {
        Set<String> brokerAddressSet = new HashSet<>();
        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
        final BrokerData brokerData = clusterInfoSerializeWrapper.getBrokerAddrTable().get(brokerName);
        if (brokerData != null) {
            brokerAddressSet.addAll(brokerData.getBrokerAddrs().values());
        }
        return brokerAddressSet;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/15
     * @方法描述：根据Broker集群名称得到该集群下所有Broker名称的方法
     */
    public static Set<String> fetchBrokerNameByClusterName(final MQAdminExt adminExt, final String clusterName)
            throws Exception {
        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
        Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
        if (brokerNameSet == null || brokerNameSet.isEmpty()) {
            throw new Exception(ERROR_MESSAGE);
        }
        return brokerNameSet;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/15
     * @方法描述：根据Broker节点地址获取Broker名称的方法
     */
    public static String fetchBrokerNameByAddr(final MQAdminExt adminExt, final String addr) throws Exception {
        //得到Broker集群的全部信息
        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
        Map<String/* brokerName */, BrokerData> brokerAddrTable = clusterInfoSerializeWrapper.getBrokerAddrTable();
        Iterator<Map.Entry<String, BrokerData>> it = brokerAddrTable.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<String, BrokerData> entry = it.next();
            HashMap<Long, String> brokerAddrs = entry.getValue().getBrokerAddrs();
            if (brokerAddrs.containsValue(addr)) {
                return entry.getKey();
            }
        }
        throw new Exception(ERROR_MESSAGE);
    }
}
